package com.org.oracle.consumer;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.util.concurrent.RateLimiter;
import com.org.oracle.constant.MQConstants;
import com.org.oracle.constant.RedisKeyConstants;
import com.org.oracle.enums.FollowUnfollowTypeEnum;
import com.org.oracle.mapper.FollowingMapper;
import com.org.oracle.mysql.dto.CountFollowUnfollowMqDTO;
import com.org.oracle.mysql.dto.FollowUserMqDTO;
import com.org.oracle.mysql.dto.UnfollowUserMqDTO;
import com.org.oracle.mysql.entity.Following;
import com.org.oracle.util.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Objects;

/**
 * @author: heiye
 * @date: 2024/12/20 下午2:03
 * @version: v1.0.0
 * @description: 关注、取消关注MQ消费者
 */
@Slf4j
@Component
@RocketMQMessageListener(
        // Group 组
        consumerGroup = "oracle_group_" + MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW,
        // 消费的主题 Topic
        topic = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW,
        // 设置为顺序消费模式
        consumeMode = ConsumeMode.ORDERLY
)
public class FollowUnfollowUserConsumer implements RocketMQListener<Message> {

    @Resource
    private FollowingMapper followingMapper;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    // 每秒创建 5000 个令牌
    private RateLimiter rateLimiter = RateLimiter.create(5000);

    @Override
    public void onMessage(Message message) {
        // 流量削峰：通过获取令牌，如果没有令牌可用，将阻塞，直到获得
        rateLimiter.acquire();

        // 消息体
        String bodyJsonStr = new String(message.getBody());
        // 标签
        String tags = message.getTags();

        log.info("==> FollowUnfollowConsumer 消费了消息 {}, tags: {}", bodyJsonStr, tags);

        // 根据 MQ 标签，判断操作类型
        // 关注
        if (Objects.equals(tags, MQConstants.TAG_FOLLOW)) {
            handleFollowTagMessage(bodyJsonStr);
        }
        // 取关
        else if (Objects.equals(tags, MQConstants.TAG_UNFOLLOW)) {
            handleUnfollowTagMessage(bodyJsonStr);
        }
    }

    /**
     * 关注
     *
     * @param bodyJsonStr
     */
    private void handleFollowTagMessage(String bodyJsonStr) {
        // 将消息体 Json 字符串转为 DTO 对象
        FollowUserMqDTO followUserMqDTO = JsonUtils.parseObject(bodyJsonStr, FollowUserMqDTO.class);

        // 判空
        if (Objects.isNull(followUserMqDTO)) {
            return;
        }

        // 幂等性：通过联合唯一索引保证

        String userId = followUserMqDTO.getUserId();
        String followingUserId = followUserMqDTO.getFollowingUserId();
        LocalDateTime createTime = followUserMqDTO.getCreateTime();

        // 构建对象
        Following following = Following.builder()
                .userId(userId)
                .followingUserId(followingUserId)
                .createTime(createTime)
                .build();
        int count = followingMapper.insert(following);

        if (count > 0) {
            // 发送 MQ 通知计数服务：统计关注数
            // 构建消息体 DTO
            CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                    .userId(userId)
                    .targetUserId(followingUserId)
                    // 关注
                    .type(FollowUnfollowTypeEnum.FOLLOW.getCode())
                    .build();

            // 发送 MQ
            sendMQ(countFollowUnfollowMqDTO);
        }
    }

    /**
     * 取关
     *
     * @param bodyJsonStr
     */
    private void handleUnfollowTagMessage(String bodyJsonStr) {
        // 将消息体 Json 字符串转为 DTO 对象
        UnfollowUserMqDTO unfollowUserMqDTO = JsonUtils.parseObject(bodyJsonStr, UnfollowUserMqDTO.class);

        // 判空
        if (Objects.isNull(unfollowUserMqDTO)) {
            return;
        }

        // 幂等性：通过联合唯一索引保证

        String userId = unfollowUserMqDTO.getUserId();
        String unfollowingUserId = unfollowUserMqDTO.getUnfollowingUserId();

        // 删除用户关注记录
        LambdaQueryWrapper<Following> followingLambdaQueryWrapper = Wrappers.<Following>lambdaQuery()
                .eq(Following::getUserId, userId)
                .eq(Following::getFollowingUserId, unfollowingUserId);
        int count = followingMapper.delete(followingLambdaQueryWrapper);

        if (count > 0) {
            // 发送 MQ 通知计数服务：统计关注数
            // 构建消息体 DTO
            CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                    .userId(userId)
                    .targetUserId(unfollowingUserId)
                    // 取关
                    .type(FollowUnfollowTypeEnum.UNFOLLOW.getCode())
                    .build();

            // 发送 MQ
            sendMQ(countFollowUnfollowMqDTO);
        }
    }


    /**
     * 发送 MQ 通知计数服务
     *
     * @param countFollowUnfollowMqDTO
     */
    private void sendMQ(CountFollowUnfollowMqDTO countFollowUnfollowMqDTO) {
        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countFollowUnfollowMqDTO)).build();

        // 异步发送 MQ 消息
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FANS, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：粉丝数】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：粉丝数】MQ 发送异常: ", throwable);
            }
        });
    }
}
